package com.hejie.service.notification.service.impl;

import com.hejie.common.exception.BusinessException;
import com.hejie.common.exception.ErrorCode;
import com.hejie.common.model.BaseResponse;
import com.hejie.service.notification.dto.NotificationDTO;
import com.hejie.service.notification.dto.NotificationRequest;
import com.hejie.service.notification.entity.Notification;
import com.hejie.service.notification.repository.NotificationRepository;
import com.hejie.service.notification.service.NotificationService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.mail.javamail.MimeMessageHelper;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.thymeleaf.TemplateEngine;
import org.thymeleaf.context.Context;

import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor
@Slf4j
public class NotificationServiceImpl implements NotificationService {

    private final NotificationRepository notificationRepository;
    private final RedisTemplate<String, Object> redisTemplate;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final JavaMailSender mailSender;
    private final TemplateEngine templateEngine;

    // Store WebSocket sessions for real-time notifications
    private final Map<Long, WebSocketSession> userWebSocketSessions = new ConcurrentHashMap<>();

    // Redis key prefixes
    private static final String REDIS_UNREAD_COUNT_PREFIX = "notification:unread_count:";
    private static final String REDIS_USER_NOTIFICATIONS_PREFIX = "notification:user:";

    // Kafka topic
    private static final String KAFKA_NOTIFICATION_TOPIC = "notification-events";

    @Override
    @Transactional
    public BaseResponse<NotificationDTO> createNotification(NotificationRequest notificationRequest) {
        // Create notification entity
        Notification notification = Notification.builder()
                .userId(notificationRequest.getUserId())
                .type(Notification.NotificationType.valueOf(notificationRequest.getType()))
                .title(notificationRequest.getTitle())
                .content(notificationRequest.getContent())
                .relatedId(notificationRequest.getRelatedId())
                .relatedType(notificationRequest.getRelatedType())
                .isRead(false)
                .isDeleted(false)
                .metadata(notificationRequest.getMetadata())
                .build();

        Notification savedNotification = notificationRepository.save(notification);
        NotificationDTO notificationDTO = convertToDTO(savedNotification);

        // Update unread count in Redis
        String unreadCountKey = REDIS_UNREAD_COUNT_PREFIX + notification.getUserId();
        redisTemplate.opsForValue().increment(unreadCountKey);

        // Clear user notifications cache
        String userNotificationsKey = REDIS_USER_NOTIFICATIONS_PREFIX + notification.getUserId();
        redisTemplate.delete(userNotificationsKey);

        // Send to Kafka for async processing
        String notificationMessage = String.format("{\"notificationId\": %d, \"userId\": %d, \"type\": \"%s\"}",
                savedNotification.getId(), notification.getUserId(), notification.getType());
        kafkaTemplate.send(KAFKA_NOTIFICATION_TOPIC, "NOTIFICATION_CREATED", notificationMessage);

        // Send real-time notification via WebSocket if user is online
        sendWebSocketNotification(notificationDTO);

        log.info("Notification created: {} for user {}", savedNotification.getId(), notification.getUserId());

        return BaseResponse.success(notificationDTO);
    }

    @Override
    public BaseResponse<Page<NotificationDTO>> getUserNotifications(Long userId, Pageable pageable) {
        // Try to get from cache first
        String cacheKey = REDIS_USER_NOTIFICATIONS_PREFIX + userId + ":" + pageable.getPageNumber() + ":" + pageable.getPageSize();
        Page<NotificationDTO> cachedNotifications = (Page<NotificationDTO>) redisTemplate.opsForValue().get(cacheKey);

        if (cachedNotifications != null) {
            return BaseResponse.success(cachedNotifications);
        }

        // Get from database
        Page<Notification> notificationPage = notificationRepository.findByUserIdAndIsDeletedFalseOrderByCreatedAtDesc(userId, pageable);
        Page<NotificationDTO> notificationDTOPage = notificationPage.map(this::convertToDTO);

        // Cache for 30 minutes
        redisTemplate.opsForValue().set(cacheKey, notificationDTOPage, 30, java.util.concurrent.TimeUnit.MINUTES);

        return BaseResponse.success(notificationDTOPage);
    }

    @Override
    public BaseResponse<Long> getUnreadCount(Long userId) {
        String unreadCountKey = REDIS_UNREAD_COUNT_PREFIX + userId;

        // Try to get from cache first
        Object cachedCount = redisTemplate.opsForValue().get(unreadCountKey);
        if (cachedCount != null) {
            return BaseResponse.success((Long) cachedCount);
        }

        // If not in cache, get from database
        long unreadCount = notificationRepository.countByUserIdAndIsReadFalseAndIsDeletedFalse(userId);

        // Update cache
        redisTemplate.opsForValue().set(unreadCountKey, unreadCount);

        return BaseResponse.success(unreadCount);
    }

    @Override
    @Transactional
    public BaseResponse<Void> markAsRead(Long notificationId, Long userId) {
        // Mark as read
        int updatedRows = notificationRepository.markAsReadById(notificationId, userId);
        if (updatedRows == 0) {
            throw new BusinessException(ErrorCode.NOTIFICATION_NOT_FOUND);
        }

        // Decrement unread count in Redis
        String unreadCountKey = REDIS_UNREAD_COUNT_PREFIX + userId;
        redisTemplate.opsForValue().decrement(unreadCountKey);

        // Clear cache
        String userNotificationsKey = REDIS_USER_NOTIFICATIONS_PREFIX + userId;
        redisTemplate.delete(userNotificationsKey);

        log.info("Notification {} marked as read by user {}", notificationId, userId);

        return BaseResponse.success();
    }

    @Override
    @Transactional
    public BaseResponse<Void> markAllAsRead(Long userId) {
        int updatedRows = notificationRepository.markAllAsRead(userId);

        if (updatedRows > 0) {
            // Reset unread count in Redis
            String unreadCountKey = REDIS_UNREAD_COUNT_PREFIX + userId;
            redisTemplate.opsForValue().set(unreadCountKey, 0L);

            // Clear cache
            String userNotificationsKey = REDIS_USER_NOTIFICATIONS_PREFIX + userId;
            redisTemplate.delete(userNotificationsKey);

            log.info("User {} marked {} notifications as read", userId, updatedRows);
        }

        return BaseResponse.success();
    }

    @Override
    @Transactional
    public BaseResponse<Void> deleteNotifications(List<Long> notificationIds, Long userId) {
        if (notificationIds == null || notificationIds.isEmpty()) {
            throw new BusinessException(ErrorCode.PARAM_VALIDATION_ERROR, "通知ID列表不能为空");
        }

        int deletedRows = notificationRepository.deleteByIds(notificationIds, userId);

        if (deletedRows > 0) {
            // Clear cache
            String userNotificationsKey = REDIS_USER_NOTIFICATIONS_PREFIX + userId;
            redisTemplate.delete(userNotificationsKey);

            log.info("User {} deleted {} notifications", userId, deletedRows);
        }

        return BaseResponse.success();
    }

    @Override
    @Transactional
    public BaseResponse<Void> deleteAllNotifications(Long userId) {
        int deletedRows = notificationRepository.deleteAllByUserId(userId);

        if (deletedRows > 0) {
            // Reset unread count in Redis
            String unreadCountKey = REDIS_UNREAD_COUNT_PREFIX + userId;
            redisTemplate.delete(unreadCountKey);

            // Clear cache
            String userNotificationsKey = REDIS_USER_NOTIFICATIONS_PREFIX + userId;
            redisTemplate.delete(userNotificationsKey);

            log.info("User {} deleted all {} notifications", userId, deletedRows);
        }

        return BaseResponse.success();
    }

    @Override
    public void sendEmailNotification(NotificationDTO notificationDTO, String email) {
        try {
            MimeMessage message = mailSender.createMimeMessage();
            MimeMessageHelper helper = new MimeMessageHelper(message, true);

            helper.setTo(email);
            helper.setSubject(notificationDTO.getTitle());

            // Create email content using Thymeleaf template
            Context context = new Context();
            context.setVariable("notification", notificationDTO);
            String htmlContent = templateEngine.process("notification-email", context);

            helper.setText(htmlContent, true);

            mailSender.send(message);
            log.info("Email notification sent to {} for notification {}", email, notificationDTO.getId());
        } catch (MessagingException e) {
            log.error("Failed to send email notification: {}", e.getMessage(), e);
        }
    }

    @Override
    public void sendWebSocketNotification(NotificationDTO notificationDTO) {
        WebSocketSession session = userWebSocketSessions.get(notificationDTO.getUserId());
        if (session != null && session.isOpen()) {
            try {
                String message = String.format("{\"id\": %d, \"type\": \"%s\", \"title\": \"%s\", \"content\": \"%s\", \"createdAt\": \"%s\"}",
                        notificationDTO.getId(),
                        notificationDTO.getType(),
                        notificationDTO.getTitle(),
                        notificationDTO.getContent(),
                        notificationDTO.getCreatedAt());
                session.sendMessage(new TextMessage(message));
                log.info("WebSocket notification sent to user {} for notification {}", notificationDTO.getUserId(), notificationDTO.getId());
            } catch (Exception e) {
                log.error("Failed to send WebSocket notification: {}", e.getMessage(), e);
                // Remove invalid session
                userWebSocketSessions.remove(notificationDTO.getUserId());
            }
        }
    }

    // Add WebSocket session when user connects
    public void addWebSocketSession(Long userId, WebSocketSession session) {
        userWebSocketSessions.put(userId, session);
        log.info("User {} connected via WebSocket", userId);
    }

    // Remove WebSocket session when user disconnects
    public void removeWebSocketSession(Long userId) {
        userWebSocketSessions.remove(userId);
        log.info("User {} disconnected from WebSocket", userId);
    }

    // Convert entity to DTO
    private NotificationDTO convertToDTO(Notification notification) {
        return NotificationDTO.builder()
                .id(notification.getId())
                .userId(notification.getUserId())
                .type(notification.getType().name())
                .title(notification.getTitle())
                .content(notification.getContent())
                .relatedId(notification.getRelatedId())
                .relatedType(notification.getRelatedType())
                .isRead(notification.getIsRead())
                .metadata(notification.getMetadata())
                .createdAt(notification.getCreatedAt())
                .updatedAt(notification.getUpdatedAt())
                .build();
    }
}